package com.itc.bi.kafaka;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.itc.bi.constant.KafkaConstant;
import com.itc.bi.constant.RedisConstant;
import com.itc.bi.dto.kafka.DataSetDTO;
import com.itc.bi.entity.DataProcessEntity;
import com.itc.bi.entity.DataProcessInfoEntity;
import com.itc.bi.entity.DirectoryEntity;
import com.itc.bi.service.DataProcessInfoService;
import com.itc.bi.service.DataProcessService;
import com.itc.bi.service.DirectoryService;
import com.itc.bi.service.impl.ClickHouseJDBCService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @BelongsProject: itc-airport-cloud
 * @BelongsPackage: com.itc.bi.kafaka
 * @Author: hecaiy
 * @CreateTime: 2024-07-03  09:16
 * @Description: TODO 数据加工数据集监听  设置加工数据集全量版本更新
 * @Version: 1.0
 */
@Slf4j
@Component
public class DataSetKafkaListener {
    @Autowired
    private DataProcessService dataProcessService;
    @Autowired
    private DataProcessInfoService dataProcessInfoService;
    @Autowired
    private ClickHouseJDBCService clickHouseJDBCService;
    @Autowired
    private DirectoryService directoryService;
    @Autowired
    private RedissonClient redissonClient;

    /**
     * 数据集加工数据数据同步
     * @param record
     */
    @KafkaListener(groupId = "bi_data_set_db",topics = KafkaConstant.KAFKA_TOPIC_DATA_SET_PROCESS)
    public void dataSetListener(ConsumerRecord record) throws InterruptedException {
        String message = (String) record.value();
        DataSetDTO dataSetDTO = JSON.parseObject(message, DataSetDTO.class);
        if(ObjectUtil.isNotEmpty(dataSetDTO) && ObjectUtil.isNotEmpty(dataSetDTO.getDtId())){
            //TODO  锁操作
            dataSetProcessSync(dataSetDTO);
        }
    }

    /**
     * 解决消息堆积问题
     * @param dataSetDTO
     */
    @Async("asyncServiceExecutor")
    public void dataSetProcessSync(DataSetDTO dataSetDTO) throws InterruptedException {
        if(dataSetDTO.getType()!=null && dataSetDTO.getType().equalsIgnoreCase("edit")){
            dataSetProcessEdit(dataSetDTO.getDtId());
        }else {
            dataSetProcessUpdate(dataSetDTO.getDtId());
        }
    }

    /**
     * 更新数据加工数据信息  同时将数据插入到版本表
     * @param dtId
     */
    public void dataSetProcessUpdate(Long dtId){
        String key = RedisConstant.EDIT_KEY+dtId;
        RLock lock = redissonClient.getLock(key);
        boolean isSuccess = lock.tryLock();
        //数据加工中，直接退出
        if(!isSuccess){
            return;
        }
        try{
            DirectoryEntity directoryEntity = directoryService.getById(dtId);
            if(ObjectUtil.isNotEmpty(directoryEntity)){
                //替换加工详情中的第一个占位符
                String tableName = directoryEntity.getTableName();
                //更新数据加工信息
                List<DataProcessEntity> updateList = new ArrayList<>();
                //更新数据加工详情信息
                List<DataProcessInfoEntity> updateInfoList = new ArrayList<>();
                List<DataProcessEntity> list = dataProcessService.list(Wrappers.<DataProcessEntity>lambdaQuery()
                        .eq(DataProcessEntity::getDtId, dtId)
                        .eq(DataProcessEntity::getType, "process"));

                List<DataProcessInfoEntity> dataProcessInfoEntityListParam = dataProcessInfoService.list(Wrappers.<DataProcessInfoEntity>lambdaQuery()
                        .eq(DataProcessInfoEntity::getUnionDtId, dtId)
                        .eq(DataProcessInfoEntity::getType, "process"));
                List<Long> dpIps = dataProcessInfoEntityListParam.stream().map(p -> p.getDpId()).distinct().collect(Collectors.toList());
                if(dpIps!=null && dpIps.size()>0){
                    List<DataProcessEntity> dataProcessEntityList = dataProcessService.list(Wrappers.<DataProcessEntity>lambdaQuery()
                            .eq(DataProcessEntity::getType, "process")
                            .in(DataProcessEntity::getId, dpIps)
                            .ne(DataProcessEntity::getDtId, dtId));
                    if(dataProcessEntityList!=null && dataProcessEntityList.size()>0){
                        list.addAll(dataProcessEntityList);
                    }
                }

                List<Long> collect = list.stream().map(DataProcessEntity::getId).collect(Collectors.toList());
                if(ObjectUtil.isNotEmpty(collect)){
                    List<DataProcessInfoEntity> infoEntityList = dataProcessInfoService.list(Wrappers.<DataProcessInfoEntity>lambdaQuery()
                            .eq(DataProcessInfoEntity::getType, "process")
                            .in(DataProcessInfoEntity::getDpId, collect)
                            .orderByAsc(DataProcessInfoEntity::getOrderNum));
                    if (CollectionUtil.isNotEmpty(infoEntityList)){
                        for (DataProcessEntity entity : list){
                            String masterTable = entity.getTableName();
                            String slaveTable = entity.getVersion();
                            //清空附表数据
                            clickHouseJDBCService.emptyTable(slaveTable);
                            List<DataProcessInfoEntity> dataProcessInfoEntityList = infoEntityList.stream().filter(p -> p.getDpId().equals(entity.getId())).collect(Collectors.toList());

                            //逐步完成数据加工  最后将主表的数据进行调整
                            for (int i = 0; i < dataProcessInfoEntityList.size(); i++) {
                                DataProcessInfoEntity dataProcessInfoEntity = dataProcessInfoEntityList.get(i);
                                String parameterValues = dataProcessInfoEntity.getParameterValues();
                                if (parameterValues.contains("${tableName}")){
                                    parameterValues = parameterValues.replace("${tableName}", tableName);
                                }
                                if(parameterValues.contains("${unionTableName}")){
                                    //需先判断数据集是不是当前数据集id，当前数据集不查询数据库
                                    if(dtId.equals(dataProcessInfoEntity.getUnionDtId())){
                                        parameterValues = parameterValues.replace("${unionTableName}", tableName);
                                    }else {
                                        DirectoryEntity directory = this.directoryService.getById(dataProcessInfoEntity.getUnionDtId());
                                        parameterValues = parameterValues.replace("${unionTableName}", directory.getTableName());
                                    }
                                }
                                //逐步操作sql
                                clickHouseJDBCService.emptyTable(dataProcessInfoEntity.getProcessTable());
                                //往表中插入数据
                                clickHouseJDBCService.insertTableProcessDS(dataProcessInfoEntity.getProcessTable(), parameterValues);
                                //最后异步 将数据表名称进行修改
                                if(i==dataProcessInfoEntityList.size()-1){
                                    slaveTable = dataProcessInfoEntity.getProcessTable();
                                    dataProcessInfoEntity.setProcessTable(slaveTable);
                                    updateInfoList.add(dataProcessInfoEntity);
                                }
                            }
                            //设置目标表为 最新创建的表名称
                            entity.setTableName(slaveTable);
                            entity.setVersion(masterTable);
                            updateList.add(entity);
                        }
                    }
                    dataProcessService.updateBatchById(updateList);
                    dataProcessInfoService.updateBatchById(updateInfoList);
                }
            }
        }finally {
            lock.unlock();
        }
    }

    /**
     * 逐步加工，新建表执行
     * @param dtId
     */
    public void dataSetProcessEdit(Long dtId) throws InterruptedException {
        String key = RedisConstant.EDIT_KEY+dtId;
        RLock lock = redissonClient.getLock(key);
        //TODO 锁机制待测试
        lock.tryLock(5,120, TimeUnit.MINUTES);
        try{
            DirectoryEntity directoryEntity = directoryService.getById(dtId);
            if(ObjectUtil.isNotEmpty(directoryEntity)){
                //替换加工详情中的第一个占位符
                String tableName = directoryEntity.getTableName();
                //更新数据加工信息
                List<DataProcessEntity> updateList = new ArrayList<>();
                //更新数据加工详情信息
                List<DataProcessInfoEntity> updateInfoList = new ArrayList<>();
                List<DataProcessEntity> list = dataProcessService.list(Wrappers.<DataProcessEntity>lambdaQuery()
                        .eq(DataProcessEntity::getDtId, dtId)
                        .eq(DataProcessEntity::getType, "process"));
                //调整
                List<DataProcessInfoEntity> dataProcessInfoEntityListParam = dataProcessInfoService.list(Wrappers.<DataProcessInfoEntity>lambdaQuery()
                        .eq(DataProcessInfoEntity::getType, "process")
                        .eq(DataProcessInfoEntity::getUnionDtId, dtId));
                List<Long> dpIps = dataProcessInfoEntityListParam.stream().map(p -> p.getDpId()).distinct().collect(Collectors.toList());
                if(dpIps!=null && dpIps.size()>0){
                    List<DataProcessEntity> dataProcessEntityList = dataProcessService.list(Wrappers.<DataProcessEntity>lambdaQuery()
                            .eq(DataProcessEntity::getType, "process")
                            .in(DataProcessEntity::getId, dpIps)
                            .ne(DataProcessEntity::getDtId, dtId));
                    if(dataProcessEntityList!=null && dataProcessEntityList.size()>0){
                        list.addAll(dataProcessEntityList);
                    }
                }

                List<Long> collect = list.stream().map(DataProcessEntity::getId).collect(Collectors.toList());
                if(ObjectUtil.isNotEmpty(collect)){
                    List<DataProcessInfoEntity> infoEntityList = dataProcessInfoService.list(Wrappers.<DataProcessInfoEntity>lambdaQuery()
                            .eq(DataProcessInfoEntity::getType, "process")
                            .in(DataProcessInfoEntity::getDpId, collect)
                            .orderByAsc(DataProcessInfoEntity::getOrderNum));
                    if (CollectionUtil.isNotEmpty(infoEntityList)){
                        for (DataProcessEntity entity : list){
                            String masterTable = "";
                            String slaveTable = "";
                            List<DataProcessInfoEntity> dataProcessInfoEntityList = infoEntityList.stream().filter(p -> p.getDpId().equals(entity.getId())).collect(Collectors.toList());
                            //需要删除的表信息
                            List<String> sourceList = new ArrayList<>();
                            //逐步完成数据加工  最后将主表的数据进行调整
                            for (int i = 0; i < dataProcessInfoEntityList.size(); i++) {
                                DataProcessInfoEntity dataProcessInfoEntity = dataProcessInfoEntityList.get(i);
                                String parameterValues = dataProcessInfoEntity.getParameterValues();
                                if (parameterValues.contains("${tableName}")){
                                    parameterValues = parameterValues.replace("${tableName}", tableName);
                                }
                                if(parameterValues.contains("${unionTableName}")){
                                    //需先判断数据集是不是当前数据集id，当前数据集不查询数据库
                                    if(dtId.equals(dataProcessInfoEntity.getUnionDtId())){
                                        parameterValues = parameterValues.replace("${unionTableName}", tableName);
                                    }else {
                                        DirectoryEntity directory = this.directoryService.getById(dataProcessInfoEntity.getUnionDtId());
                                        parameterValues = parameterValues.replace("${unionTableName}", directory.getTableName());
                                    }
                                }
                                //逐步操作sql
                                sourceList.add(dataProcessInfoEntity.getProcessTable());
                                String processTable = dataProcessInfoEntity.getProcessTable();
                                processTable = processTable.substring(0,processTable.lastIndexOf("_"))+IdWorker.getId();
                                //往表中插入数据
                                clickHouseJDBCService.createTableProcessDS(processTable,null, parameterValues);
                                //最后一步 将数据表名称进行修改
                                if(i==dataProcessInfoEntityList.size()-1){
                                    masterTable = processTable;
                                    //创建附表
                                    String slaveProcessTable = processTable.substring(0,processTable.lastIndexOf("_"))+"_"+IdWorker.getId();
                                    slaveTable = slaveProcessTable;
                                    clickHouseJDBCService.createTableProcessDS(masterTable,null, parameterValues);
                                }
                                dataProcessInfoEntity.setProcessTable(processTable);
                                updateInfoList.add(dataProcessInfoEntity);
                            }
                            //设置目标表为 最新创建的表名称
                            entity.setTableName(masterTable);
                            entity.setVersion(slaveTable);
                            updateList.add(entity);
                            if(ObjectUtil.isNotEmpty(sourceList)){
                                for (String tableNameParam : sourceList){
                                    //删除表
                                    clickHouseJDBCService.dropTable(tableNameParam);
                                }
                            }
                        }
                    }
                    dataProcessService.updateBatchById(updateList);
                    dataProcessInfoService.updateBatchById(updateInfoList);
                }
            }
        }finally {
            lock.unlock();
        }
    }

    public void dataSetProcess(Long dtId){
        List<DataProcessEntity> updateList = new ArrayList<>();
        List<DataProcessEntity> list = dataProcessService.list(Wrappers.<DataProcessEntity>lambdaQuery().eq(DataProcessEntity::getDtId, dtId));
        List<Long> collect = list.stream().map(DataProcessEntity::getId).collect(Collectors.toList());
        if(ObjectUtil.isNotEmpty(collect)){
            List<DataProcessInfoEntity> infoEntityList = dataProcessInfoService.list(Wrappers.<DataProcessInfoEntity>lambdaQuery()
                    .in(DataProcessInfoEntity::getDpId, collect));
            if (CollectionUtil.isNotEmpty(infoEntityList)){
                for (DataProcessEntity entity : list){
                    //在数据操作完之后  把临时表删除
                    List<String> tableNameList = new ArrayList<>();
                    //数据操作表前缀 pro_dtId_随机码
                    String tableNameParam = "pro_"+dtId+"_";
                    String masterTable = entity.getTableName();
                    List<DataProcessInfoEntity> dataProcessInfoEntityList = infoEntityList.stream().filter(p -> p.getDpId().equals(entity.getId()))
                            .sorted(Comparator.comparing(DataProcessInfoEntity::getOrderNum, Comparator.reverseOrder())).collect(Collectors.toList());

                    //逐步完成数据加工  最后将主表的数据进行调整
                    for (int i = 0; i < dataProcessInfoEntityList.size(); i++) {
                        Long idParam = IdWorker.getId();
                        tableNameParam = tableNameParam + idParam;
                        DataProcessInfoEntity dataProcessInfoEntity = dataProcessInfoEntityList.get(i);
                        if (ObjectUtil.isNotEmpty(dataProcessInfoEntity.getId())){
                            //逐步操作sql
                            String sql = dataProcessInfoEntity.getParameterValues();
                            //建表并且同步数据
                            clickHouseJDBCService.createTableProcessDS(tableNameParam, dtId, sql);
                            //最后一次加工的数据表不删除
                            if(i!=dataProcessInfoEntityList.size()-1){
                                tableNameList.add(tableNameParam);
                            }
                        }
                    }

                    if(ObjectUtil.isNotEmpty(tableNameList)){
                        for (String tableName : tableNameList){
                            //删除表
                            clickHouseJDBCService.dropTable(tableName);
                        }
                    }
                    //设置目标表为 最新创建的字词
                    entity.setTableName(tableNameParam);
                    entity.setVersion(masterTable);
                    updateList.add(entity);
                }
            }
            dataProcessService.updateBatchById(updateList);
        }
    }
}
